package com.ruoze.bdp.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

public class HBaseUtil {


    public static Optional<Connection> getHbaseConnection(String zkHost, Integer zkPort){
        try{
            Configuration config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", zkHost);
            config.set("hbase.zookeeper.property.clientPort", zkPort.toString());
            config.set("zookeeper.recovery.retry", "1");
            config.set("hbase.client.retries.number", "1");

            Connection connection = ConnectionFactory.createConnection(config);
            connection.getAdmin().listTableNames();
            return Optional.of(connection);
        }catch (Exception e){
            return Optional.empty();
        }
    }

    public static boolean testConnection(String zkHost, Integer zkPort) {
        try {
            Optional<Connection> connectionOpt = getHbaseConnection(zkHost, zkPort);
            if (connectionOpt.isPresent()) {
                connectionOpt.get().close();
                return true;
            } else {
                return false;
            }
        } catch (Exception e) {
            return false;
        }
    }

    public static List<String> getNamespaces(Connection connection){
        try {
            return Arrays.stream(connection.getAdmin().listNamespaceDescriptors()).map(i -> i.getName()).collect(Collectors.toList());
        } catch (Exception e) {
            return null;
        }
    }

    public static List<String> getTables(Connection connection, String namespace){
        try {
            if(namespace == null || namespace.isEmpty()){
                return Arrays.stream(connection.getAdmin().listTableNames()).map(i -> i.getNameAsString()).collect(Collectors.toList());
            }else{
                return Arrays.stream(connection.getAdmin().listTableNamesByNamespace(namespace))
                        .map(i -> i.getNameAsString())
                        .map(i -> {
                            String[] split = i.split(":");
                            if(split.length == 1){
                                return split[0];
                            }else{
                                return split[1];
                            }
                        })
                        .collect(Collectors.toList());
            }
        } catch (Exception e) {
            return null;
        }
    }

    public static List<String> getFamilyColumns(Connection connection, String namespace, String tableName){
        try {
            HTableDescriptor tableDescriptor = connection.getAdmin().getTableDescriptor(TableName.valueOf(namespace + ":" + tableName));
            return Arrays.stream(tableDescriptor.getColumnFamilies()).map(i -> i.getNameAsString()).collect(Collectors.toList());
        } catch (Exception e) {
            return null;
        }
    }

    public static boolean createNamespace(Connection connection, String namespaceName) {
        try {
                connection.getAdmin().createNamespace(NamespaceDescriptor.create(namespaceName).build());
                return true;
        } catch (Exception e) {
            return false;
        }
    }

    public static boolean createTable(Connection connection, String namespaceName, String tableName, List<String> cfs){
        try {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(namespaceName + ":" + tableName);
            for(String cf : cfs){
                hTableDescriptor.addFamily(new org.apache.hadoop.hbase.HColumnDescriptor(cf));
            }
            connection.getAdmin().createTable(hTableDescriptor);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    public static String getRowKey(String heading, String separator, List<String> cols, Map<String, Map<String, String>> values){
        StringBuilder sb = new StringBuilder(heading);
        sb.append(separator);

        for(int i = 0; i < cols.size(); i++){
            String col = cols.get(i);
            String[] colArr = col.split("\\.");

            String cf = colArr[0];
            String column = colArr[1];

            String value = "";
            if(values.containsKey(cf) && values.get(cf).containsKey(column)){
                value = values.get(cf).get(column);
            }

            sb.append(value);

            if(i != cols.size() - 1){
                sb.append(separator);
            }
        }

        return sb.toString();
    }
}
